package com.googlecode.objectify.hbase;

import com.google.appengine.api.datastore.Blob;
import com.google.appengine.api.datastore.Category;
import com.google.appengine.api.datastore.DataTypeUtils;
import com.google.appengine.api.datastore.Email;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.GeoPt;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Link;
import com.google.appengine.api.datastore.PhoneNumber;
import com.google.appengine.api.datastore.PostalAddress;
import com.google.appengine.api.datastore.Rating;
import com.google.appengine.api.datastore.ShortBlob;
import com.google.appengine.api.datastore.Text;
import com.google.appengine.api.users.User;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.Map;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author Secondary User
 */
public class HBaseTranslator
{

	private static final byte SHORT_BLOB = 'b';
	private static final byte BLOB = 'B';
	private static final byte STRING = 'S';
	private static final byte INT = 'i';
	private static final byte LONG = 'l';
	private static final byte SHORT = 's';
	private static final byte FLOAT = 'f';
	private static final byte DOUBLE = 'd';
	private static final byte BOOLEAN_TRUE = 't';
	private static final byte BOOLEAN_FALSE = 'F';
	private static final byte TEXT = 'T';
	private static final byte DATE = 'D';
	private static final byte GEO_PT = 'G';
	private static final byte POSTAL = 'P';
	private static final byte PHONE = 'p';
	private static final byte EMAIL = 'E';
	private static final byte USER = 'U';
	private static final byte IM_HANDLE = 'I';
	private static final byte LINK = 'L';
	private static final byte CATEGORY = 'C';
	private static final byte RATING = 'R';
	private static final byte KEY = 'K';
	private static final byte BLOB_KEY = 'k';
	private static final byte EMBEDDED = 'e';
	private static final byte NULL = 'n';

	public static Entity toEntity(Map<byte[], byte[]> map, byte[] row)
	{
		Key key = Key.fromKeyString(new String(row));
		Entity toReturn = new Entity(key);
		for (Map.Entry<byte[], byte[]> entry : map.entrySet())
		{
			String name = new String(entry.getKey());
			Object value = toObject(entry.getValue());
			toReturn.setProperty(name, value);
		}
		return toReturn;
	}

	public static Object toObject(byte[] value)
	{
		if (value.length == 0)
		{
			return null;
		}
		int length = value.length - 1;
		switch (value[value.length - 1])
		{
			case SHORT_BLOB:
				if (length <= DataTypeUtils.MAX_SHORT_BLOB_PROPERTY_LENGTH)
				{
					byte[] bytes = new byte[length];
					System.arraycopy(bytes, 0, value, 0, length);
					return new ShortBlob(bytes);
				}
				else
				{
					return null;//should never happen
				}
			case BLOB:
			{
				byte[] bytes = new byte[length];
				System.arraycopy(bytes, 0, value, 0, length);
				return new Blob(bytes);
			}
			case STRING:
				return Bytes.toString(value, 0, length);
			case INT:
				return Bytes.toInt(value);
			case LONG:
				return Bytes.toLong(value);
			case SHORT:
				return Bytes.toShort(value);
			case FLOAT:
				return Bytes.toFloat(value);
			case DOUBLE:
				return Bytes.toDouble(value);
			case BOOLEAN_TRUE:
				return true;
			case BOOLEAN_FALSE:
				return false;
			case TEXT:
				return new Text(Bytes.toString(value, 0, length));
			case DATE:
				return new Date(Bytes.toLong(value));
			case GEO_PT:
				return new GeoPt(Bytes.toFloat(value, 0), Bytes.toFloat(value, Bytes.SIZEOF_FLOAT));
			case POSTAL:
				return new PostalAddress(Bytes.toString(value, 0, length));
			case PHONE:
				return new PhoneNumber(Bytes.toString(value, 0, length));
			case EMAIL:
				return new Email(Bytes.toString(value, 0, length));
			case USER:
				return User.getUser(Bytes.toString(value, 0, length));
			case IM_HANDLE:
				return null;//TODO: new IMHandle(Bytes.toString(value, 0, length));
			case LINK:
				return new Link(Bytes.toString(value, 0, length));
			case CATEGORY:
				return new Category(Bytes.toString(value, 0, length));
			case RATING:
				return new Rating(Bytes.toInt(value));
			case KEY:
				return Key.fromKeyString(Bytes.toString(value, 0, length));
			case BLOB_KEY:
				return Bytes.toString(value, 0, length);
			case EMBEDDED:
				return null; //TODO: Code the embedded entity type.
			case NULL:
				return null;
		}
		throw new IllegalArgumentException("TYpe of object not recognised.");
	}

	public static byte[] toBytes(Object val)
	{
		byte[][] toReturn = new byte[1][];
		if (internalToBytes(val, toReturn, true))
		{
			return toReturn[0];
		}
		return null;
	}

	private static boolean internalToBytes(Object val, byte[][] output, boolean resultNeeded)
	{
		if (val == null)
		{
			if (resultNeeded)
			{
				output[0] = new byte[1];
				output[0][0] = NULL;
			}
			return true;
		}
		if (val instanceof String)
		{
			if (resultNeeded)
			{
				String valS = (String) val;
				output[0] = new byte[(valS.length() * Bytes.SIZEOF_CHAR) + 1];
				for (int i = 0; i < valS.length(); i++)
				{
					int arrayLoc = i * Bytes.SIZEOF_CHAR;
					char c = valS.charAt(i);
					output[0][arrayLoc + 1] = (byte) c;
					c >>= 8;
					output[0][arrayLoc] = (byte) c;
				}
				output[0][output[0].length - 1] = STRING;
				//replace this lot with ByteBuffer??
			}
			return true;
		}
		if (val instanceof Integer)
		{
			if (resultNeeded)
			{
				output[0] = new byte[Bytes.SIZEOF_INT + 1];
				ByteBuffer.wrap(output[0]).putInt((Integer) val).put(INT);
			}
			return true;
		}
		if (val instanceof Float)
		{
			if (resultNeeded)
			{
				output[0] = new byte[Bytes.SIZEOF_FLOAT + 1];
				ByteBuffer.wrap(output[0]).putFloat((Float) val).put(FLOAT);
			}
			return true;
		}
		if (val instanceof Double)
		{
			if (resultNeeded)
			{
				output[0] = new byte[Bytes.SIZEOF_DOUBLE + 1];
				ByteBuffer.wrap(output[0]).putDouble((Double) val).put(DOUBLE);
			}
			return true;
		}
		if (val instanceof Long)
		{
			if (resultNeeded)
			{
				output[0] = new byte[Bytes.SIZEOF_LONG + 1];
				ByteBuffer.wrap(output[0]).putLong((Long) val).put(LONG);
			}
			return true;
		}
		if (val instanceof Short)
		{
			if (resultNeeded)
			{
				output[0] = new byte[Bytes.SIZEOF_SHORT + 1];
				ByteBuffer.wrap(output[0]).putShort((Short) val).put(SHORT);
			}
			return true;
		}
		if (val instanceof Boolean)
		{
			if (resultNeeded)
			{
				output[0] = new byte[1];
				if ((Boolean) val)
				{
					output[0][0] = BOOLEAN_TRUE;
				}
				else
				{
					output[0][0] = BOOLEAN_FALSE;
				}
			}
			return true;
		}
		if (val instanceof ShortBlob)
		{
			ShortBlob blob = (ShortBlob) val;
			if (blob.getBytes().length > DataTypeUtils.MAX_SHORT_BLOB_PROPERTY_LENGTH)
			{
				return false;
			}
			if (resultNeeded)
			{
				output[0] = new byte[blob.getBytes().length + 1];
				ByteBuffer.wrap(output[0]).put(blob.getBytes()).put(SHORT_BLOB);
			}
			return true;
		}
		if (val instanceof Blob)
		{
			Blob blob = (Blob) val;
			if (resultNeeded)
			{
				output[0] = new byte[blob.getBytes().length + 1];
				ByteBuffer.wrap(output[0]).put(blob.getBytes()).put(BLOB);
			}
			return true;
		}
		if (val instanceof Text)
		{
			if (resultNeeded)
			{
				internalToBytes(((Text) val).getValue(), output, true);
				output[0][output[0].length - 1] = TEXT;
			}
			return true;
		}
		if (val instanceof Date)
		{
			if (resultNeeded)
			{
				internalToBytes(((Date)val).getTime(), output, true);
				output[0][output[0].length - 1] = DATE;
			}
			return true;
		}
		if (val instanceof GeoPt)
		{
			if (resultNeeded)
			{
				GeoPt g = (GeoPt)val;
				output[0] = new byte[Bytes.SIZEOF_FLOAT + Bytes.SIZEOF_FLOAT + 1];
				ByteBuffer.wrap(output[0]).putFloat(g.getLatitude()).putFloat(g.getLongitude()).put(GEO_PT);
			}
			return true;
		}
		if (val instanceof Key)
		{
			if (resultNeeded)
			{
				internalToBytes(((Key)val).toKeyString(), output, true);
				output[0][output[0].length - 1] = KEY;
			}
			return true;
		}

//	TODO: POSTAL
//	TODO: PHONE
//	TODO: EMAIL
//	TODO: USER
//	TODO: IM_HANDLE
//	TODO: LINK
//	TODO: CATEGORY
//	TODO: RATING
//	TODO: BLOB_KEY
//	TODO: EMBEDDED

		return false;
	}

	public static boolean isStorable(Object val)
	{
		return internalToBytes(val, null, false);
	}
}
